package com.twm.boot;

import com.twm.client.CuratorClient;
import com.twm.client.RPCConsumer;
import com.twm.entity.RpcRequest;
import com.twm.service.IUserService;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

public class ConsumerBoot {

    public static final Map<Integer, ServerMapper> serverMapper = new ConcurrentHashMap<Integer, ServerMapper>();

    public static void main(String[] args) throws Exception {

        //获取连接
        CuratorFramework client = CuratorClient.getCuratorClient();
        //获取节点
        String path = CuratorClient.path;
        //监听逻辑
        zkWatch(client, path);
    }

    private static void runServer() {
        AtomicReference<IUserService> iUserService = new AtomicReference<>();
        final Long[] time = {0L};
        serverMapper.forEach((k, v) -> {
            if (System.currentTimeMillis() - v.getLastTime() > 5000){
                return;
            }
            if (v.getLastTime() > time[0]){
                iUserService.set(v.getiUserService());
                time[0] = v.getLastTime();
            }
        });
        if (iUserService.get() != null){
            //循环给服务 器写数据
            RpcRequest rpcRequest = new RpcRequest();
            rpcRequest.setClassName("com.ghc.entity.RpcRequest");
            rpcRequest.setMethodName("getRpcRequest");
            rpcRequest.setRequestId("userServiceImpl");
            rpcRequest.setParameters(new Object[]{"Hello------"});
            rpcRequest.setParameterTypes(new Class[]{String.class});
            String result = iUserService.get().getRpcRequest(rpcRequest);
            System.out.println(result);
        }

    }

    /**
     * 注册监听
     * TreeCache: 可以将指定的路径节点作为根节点（祖先节点），对其所有的子节点操作进行监听，
     * 呈现树形目录的监听，可以设置监听深度，最大监听深度为 int 类型的最大值。
     */
    private static void zkWatch(final CuratorFramework client, final String path) throws Exception {
        TreeCache treeCache = new TreeCache(client, path);

        treeCache.getListenable().addListener(new TreeCacheListener() {
            public void childEvent(CuratorFramework client, TreeCacheEvent event) {
                ChildData eventData = event.getData();
                switch (event.getType()) {
                    case NODE_ADDED:
                        System.out.println(path + "节点添加" + eventData.getPath() + "\t添加数据为：" + new String(eventData.getData()));
                        String path1 = eventData.getPath();
                        String time = new String(eventData.getData());
                        final int port = Integer.parseInt(path1.substring(path1.lastIndexOf("/") + 1));
                        //创建代理对象
                        IUserService service = (IUserService) RPCConsumer.createProxy(IUserService.class, port, serverMapper);
                        if (serverMapper.get(port) == null){
                            ServerMapper mapper = new ServerMapper();
                            mapper.setiUserService(service);
                            mapper.setLastTime(Long.valueOf(time));
                            serverMapper.put(port, mapper);
                        }
                        /*Thread thread = new Thread(new Runnable() {
                            public void run() {
                                while (serverMapper.get(port) != null){
                                    runServer(port);
                                    try {
                                        Thread.sleep(3000);
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                    }
                                }
                            }
                        });
                        thread.start();*/
                        break;
                    case NODE_UPDATED:
                        System.out.println(eventData.getPath() + "节点数据更新\t更新数据为：" + new String(eventData.getData()) + "\t版本为：" + eventData.getStat().getVersion());
                        String path = eventData.getPath();
                        String time1 = new String(eventData.getData());
                        final int port1 = Integer.parseInt(path.substring(path.lastIndexOf("/") + 1));
                        serverMapper.get(port1).setLastTime(Long.valueOf(time1));
                        break;
                    case NODE_REMOVED:
                        System.out.println(eventData.getPath() + "节点被删除");
                        String path2 = eventData.getPath();
                        int i = Integer.parseInt(path2.substring(path2.lastIndexOf("/") + 1));
                        ConsumerBoot.serverMapper.remove(i);
                        break;
                    default:
                        break;
                }
            }
        });
        treeCache.start();

        while (true){
            runServer();
            Thread.sleep(3000);
        }

//        Thread.sleep(Integer.MAX_VALUE);
    }
}
